/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.consumer;

import com.alibaba.rocketmq.client.QueryResult;
import com.alibaba.rocketmq.client.Validators;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.PullCallback;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.client.consumer.listener.MessageListener;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import com.alibaba.rocketmq.client.consumer.rebalance.AllocateMessageQueueByConfig;
import com.alibaba.rocketmq.client.consumer.rebalance.AllocateMessageQueueByMachineRoom;
import com.alibaba.rocketmq.client.consumer.store.LocalFileOffsetStore;
import com.alibaba.rocketmq.client.consumer.store.OffsetStore;
import com.alibaba.rocketmq.client.consumer.store.ReadOffsetType;
import com.alibaba.rocketmq.client.consumer.store.RemoteBrokerOffsetStore;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.hook.ConsumeMessageContext;
import com.alibaba.rocketmq.client.hook.ConsumeMessageHook;
import com.alibaba.rocketmq.client.hook.FilterMessageHook;
import com.alibaba.rocketmq.client.impl.CommunicationMode;
import com.alibaba.rocketmq.client.impl.MQClientManager;
import com.alibaba.rocketmq.client.impl.factory.MQClientInstance;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.client.stat.ConsumerStatsManager;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.ServiceState;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.filter.FilterAPI;
import com.alibaba.rocketmq.common.help.FAQUrl;
import com.alibaba.rocketmq.common.message.*;
import com.alibaba.rocketmq.common.protocol.body.ConsumeStatus;
import com.alibaba.rocketmq.common.protocol.body.ConsumerRunningInfo;
import com.alibaba.rocketmq.common.protocol.body.ProcessQueueInfo;
import com.alibaba.rocketmq.common.protocol.body.QueueTimeSpan;
import com.alibaba.rocketmq.common.protocol.heartbeat.ConsumeType;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
import com.alibaba.rocketmq.common.protocol.route.BrokerData;
import com.alibaba.rocketmq.common.protocol.route.TopicRouteData;
import com.alibaba.rocketmq.common.sysflag.PullSysFlag;
import com.alibaba.rocketmq.remoting.RPCHook;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

import org.slf4j.Logger;

import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-6-15
 */
public class DefaultMQPushConsumerImpl implements MQConsumerInner {
	/**
	 * Delay some time when exception occur
	 */
	private static final long PullTimeDelayMillsWhenException = 3000;
	/**
	 * Flow control interval
	 */
	private static final long PullTimeDelayMillsWhenFlowControl = 50;
	/**
	 * Delay some time when suspend pull service
	 */
	private static final long PullTimeDelayMillsWhenSuspend = 1000;
	private static final long BrokerSuspendMaxTimeMillis = 1000 * 15;
	private static final long ConsumerTimeoutMillisWhenSuspend = 1000 * 30;
	private final Logger log = ClientLogger.getLog();
	private final DefaultMQPushConsumer defaultMQPushConsumer;

	// RebalancePushImpl：主要负责决定，当前的consumer应该从哪些Queue中消费消息；
	private final RebalanceImpl rebalanceImpl = new RebalancePushImpl(this);
	private ServiceState serviceState = ServiceState.CREATE_JUST;

	// 4）MQClientFactory：大杂烩，负责管理client（consumer、producer），并提供多种功能接口供各个Service（Rebalance、PullMessage等）调用；大部分逻辑均在这个类中完成；
	private MQClientInstance mQClientFactory;

	// 1）PullAPIWrapper：长连接，负责从broker处拉取消息，然后利用ConsumeMessageService回调用户的Listener执行消息消费逻辑；
	private PullAPIWrapper pullAPIWrapper;
	private volatile boolean pause = false;
	private boolean consumeOrderly = false;
	private MessageListener messageListenerInner;
	// 3）OffsetStore：维护当前consumer的消费记录（offset）；有两种实现，Local和Rmote，Local存储在本地磁盘上，适用于BROADCASTING广播消费模式；而Remote则将消费进度存储在Broker上，适用于CLUSTERING集群消费模式；
	private OffsetStore offsetStore;

	// 2）ConsumeMessageService：实现所谓的"Push-被动"消费机制；从Broker拉取的消息后，封装成ConsumeRequest提交给ConsumeMessageSerivce，此service负责回调用户的Listener消费消息；
	private ConsumeMessageService consumeMessageService;

	private final ArrayList<FilterMessageHook> filterMessageHookList = new ArrayList<FilterMessageHook>();

	private final long consumerStartTimestamp = System.currentTimeMillis();

	public void registerFilterMessageHook(final FilterMessageHook hook) {
		this.filterMessageHookList.add(hook);
		log.info("register FilterMessageHook Hook, {}", hook.hookName());
	}

	private final ArrayList<ConsumeMessageHook> consumeMessageHookList = new ArrayList<ConsumeMessageHook>();

	private final RPCHook rpcHook;

	public DefaultMQPushConsumerImpl(DefaultMQPushConsumer defaultMQPushConsumer, RPCHook rpcHook) {
		this.defaultMQPushConsumer = defaultMQPushConsumer;
		this.rpcHook = rpcHook;
	}

	public boolean hasHook() {
		return !this.consumeMessageHookList.isEmpty();
	}

	public void registerConsumeMessageHook(final ConsumeMessageHook hook) {
		this.consumeMessageHookList.add(hook);
		log.info("register consumeMessageHook Hook, {}", hook.hookName());
	}

	public void executeHookBefore(final ConsumeMessageContext context) {
		if (!this.consumeMessageHookList.isEmpty()) {
			for (ConsumeMessageHook hook : this.consumeMessageHookList) {
				try {
					// 调用ArrayList<ConsumeMessageHook>列表中每个ConsumeMessageHook对象的consumeMessageBefore (ConsumeMessageContext context)方法
					/**
					 * 该consumeMessageHookList变量由业务层调用DefaultMQPushConsumerImpl.registerConsumeMessageHook (ConsumeMessageHook hook)方法设置，
					 * 由业务层自定义ConsumeMessageHook接口的实现类，
					 * 实现该接口的consumeMessageBefore(final ConsumeMessageContext context)和consumeMessageAfter(final ConsumeMessageContext context)方法，
					 * 分别在回调业务层的具体消费方法之前和之后调用者两个方法。
					 */
					hook.consumeMessageBefore(context);
				} catch (Throwable e) {
				}
			}
		}
	}

	public void executeHookAfter(final ConsumeMessageContext context) {
		if (!this.consumeMessageHookList.isEmpty()) {
			for (ConsumeMessageHook hook : this.consumeMessageHookList) {
				try {
					hook.consumeMessageAfter(context);
				} catch (Throwable e) {
				}
			}
		}
	}

	public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
		createTopic(key, newTopic, queueNum, 0);
	}

	public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
		this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
	}

	public Set<MessageQueue> fetchSubscribeMessageQueues(String topic) throws MQClientException {
		Set<MessageQueue> result = this.rebalanceImpl.getTopicSubscribeInfoTable().get(topic);
		if (null == result) {
			this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
			result = this.rebalanceImpl.getTopicSubscribeInfoTable().get(topic);
		}

		if (null == result) {
			throw new MQClientException("The topic[" + topic + "] not exist", null);
		}

		return result;
	}

	public DefaultMQPushConsumer getDefaultMQPushConsumer() {
		return defaultMQPushConsumer;
	}

	public long earliestMsgStoreTime(MessageQueue mq) throws MQClientException {
		return this.mQClientFactory.getMQAdminImpl().earliestMsgStoreTime(mq);
	}

	public long maxOffset(MessageQueue mq) throws MQClientException {
		return this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
	}

	public long minOffset(MessageQueue mq) throws MQClientException {
		return this.mQClientFactory.getMQAdminImpl().minOffset(mq);
	}

	public OffsetStore getOffsetStore() {
		return offsetStore;
	}

	public void setOffsetStore(OffsetStore offsetStore) {
		this.offsetStore = offsetStore;
	}

	@Override
	public String groupName() {
		return this.defaultMQPushConsumer.getConsumerGroup();
	}

	@Override
	public MessageModel messageModel() {
		return this.defaultMQPushConsumer.getMessageModel();
	}

	@Override
	public ConsumeType consumeType() {
		return ConsumeType.CONSUME_PASSIVELY;
	}

	@Override
	public ConsumeFromWhere consumeFromWhere() {
		return this.defaultMQPushConsumer.getConsumeFromWhere();
	}

	@Override
	public Set<SubscriptionData> subscriptions() {
		Set<SubscriptionData> subSet = new HashSet<SubscriptionData>();

		subSet.addAll(this.rebalanceImpl.getSubscriptionInner().values());

		return subSet;
	}

	@Override
	public void doRebalance() {
		if (this.rebalanceImpl != null) {
			// *****************************
			this.rebalanceImpl.doRebalance();
		}
	}

	@Override
	public void persistConsumerOffset() {
		try {
			this.makeSureStateOK();
			Set<MessageQueue> mqs = new HashSet<MessageQueue>();
			// 1、获取DefaultMQPushConsumerImpl.rebalanceImpl变量的processQueueTable:ConcurrentHashMap<MessageQueue, ProcessQueue>变量值，
			// 取该变量的key值集合，即MessageQueue集合；
			Set<MessageQueue> allocateMq = this.rebalanceImpl.getProcessQueueTable().keySet(); // ProcessQueueTable()===>ConcurrentHashMap<MessageQueue, ProcessQueue>
			if (allocateMq != null) {
				mqs.addAll(allocateMq);
			}

			/**
			 * 2、若消息模式是广播（BROADCASTING），即DefaultMQPushConsumerImpl.offsetStore变量初始化为LocalFileOffsetStore对象，
			 *    在此调用LocalFileOffsetStore.persistAll(Set<MessageQueue> mqs)方法，在此方法中遍历LocalFileOffsetStore.offsetTable:ConcurrentHashMap<MessageQueue,AtomicLong>变量，
			 *    将包含在第1步的MessageQueue集合中的MessageQueue对象的消费进度持久化到consumerOffset.json物理文件中；
			 * 3、若消息模式为集群模式，即DefaultMQPushConsumerImpl.offsetStore变量初始化为RemoteBrokerOffsetStore对象，
			 *    在此调用RemoteBrokerOffsetStore.persistAll(Set<MessageQueue> mqs)方法，在此方法中遍历RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap <MessageQueue,AtomicLong>变量；
			 *    对于包含在第1步的MessageQueue集合中的MessageQueue对象，调用updateConsumeOffsetToBroker(MessageQueuemq, long offset)方法向Broker发送UPDATE_CONSUMER_OFFSET请求码的消费进度信息；
			 */
			this.offsetStore.persistAll(mqs);
		} catch (Exception e) {
			log.error("group: " + this.defaultMQPushConsumer.getConsumerGroup() + " persistConsumerOffset exception", e);
		}
	}

	@Override
	public void updateTopicSubscribeInfo(String topic, Set<MessageQueue> info) {
		// 获取DefaultMQPushConsumerImpl.rebalanceImpl:RebalanceImpl变量中的
		// RebalanceImpl.subscriptionInner：ConcurrentHashMap<String, SubscriptionData>变量值subTable；
		Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
		if (subTable != null) {
			// 若topic在该subscriptionInner变量值的列表中
			if (subTable.containsKey(topic)) {
				// 则用topic和MessageQueue集合更新RebalanceImpl. topicSubscribeInfoTable:ConcurrentHashMap<String/* topic */, Set<MessageQueue>>变量；
				this.rebalanceImpl.topicSubscribeInfoTable.put(topic, info);
			}
		}
	}

	public ConcurrentHashMap<String, SubscriptionData> getSubscriptionInner() {
		return this.rebalanceImpl.getSubscriptionInner();
	}

	@Override
	public boolean isSubscribeTopicNeedUpdate(String topic) {
		Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
		if (subTable != null) {
			if (subTable.containsKey(topic)) {
				return !this.rebalanceImpl.topicSubscribeInfoTable.containsKey(topic);
			}
		}

		return false;
	}

	private void correctTagsOffset(final PullRequest pullRequest) {
		if (0L == pullRequest.getProcessQueue().getMsgCount().get()) {
			// 若没有获取到消息（即ProcessQueue.msgCount等于0）则更新消息进度。
			// 集群
			// 广播
			this.offsetStore.updateOffset(pullRequest.getMessageQueue(), pullRequest.getNextOffset(), true);
		}
	}

	private long flowControlTimes1 = 0;
	private long flowControlTimes2 = 0;

	// 消息拉取入口
	public void pullMessage(final PullRequest pullRequest) {
		final ProcessQueue processQueue = pullRequest.getProcessQueue();

		// 1、检查PullRequest对象中的ProcessQueue对象的dropped是否为true
		// （在RebalanceService线程中为topic下的MessageQueue创建拉取消息请求时要维护对应的ProcessQueue对象，若Consumer不再订阅该topic则会将该对象的dropped置为true）
		if (processQueue.isDropped()) {
			// 若是则认为该请求是已经取消的，则直接跳出该方法；
			log.info("the pull request[{}] is droped.", pullRequest.toString());
			return;
		}

		// 2、更新PullRequest对象中的ProcessQueue对象的时间戳（ProcessQueue.lastPullTimestamp）为当前时间戳；
		pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());

		try {
			// 3、检查该Consumer是否运行中，即DefaultMQPushConsumerImpl.serviceState是否为RUNNING;
			this.makeSureStateOK();
		} catch (MQClientException e) {
			// 若不是运行状态，则调用PullMessageService.executePullRequestLater(PullRequest pullRequest, long timeDelay)方法延迟再拉取消息，其中timeDelay=3000；该方法的目的是在3秒之后再次将该PullRequest对象放入PullMessageService. pullRequestQueue队列中；并跳出该方法；
			log.warn("pullMessage exception, consumer state not ok", e);
			this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenException);
			return;
		}

		if (this.isPause()) {
			// 若是暂停状态（DefaultMQPushConsumerImpl.pause=true），则调用PullMessageService.executePullRequestLater(PullRequest pullRequest, long timeDelay)方法延迟再拉取消息，其中timeDelay=3000；该方法的目的是在3秒之后再次将该PullRequest对象放入PullMessageService. pullRequestQueue队列中；并跳出该方法；
			log.warn("consumer was paused, execute pull request later. instanceName={}", this.defaultMQPushConsumer.getInstanceName());
			this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenSuspend);
			return;
		}

		// 4、进行流控。
		long size = processQueue.getMsgCount().get();
		// 若ProcessQueue对象的msgCount大于了消费端的流控阈值（DefaultMQPushConsumer.pullThresholdForQueue，默认值为1000），
		if (size > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
			// 调用PullMessageService.executePullRequestLater方法，在50毫秒之后重新该PullRequest请求放入PullMessageService.pullRequestQueue队列中；并跳出该方法
			this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenFlowControl);
			if ((flowControlTimes1++ % 1000) == 0) {
				log.warn("the consumer message buffer is full, so do flow control, {} {} {}", size, pullRequest, flowControlTimes1);
			}
			return;
		}

		// 5、若不是顺序消费（即DefaultMQPushConsumerImpl.consumeOrderly等于false），
		if (!this.consumeOrderly) {
			// 则检查ProcessQueue对象的msgTreeMap:TreeMap<Long,MessageExt>变量的第一个key值与最后一个key值之间的差额，该key值表示查询的队列偏移量queueoffset；
			if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
				// 若差额大于阈值（由DefaultMQPushConsumer. consumeConcurrentlyMaxSpan指定，默认是2000），则调用PullMessageService.executePullRequestLater方法，在50毫秒之后重新将该PullRequest请求放入PullMessageService.pullRequestQueue队列中；并跳出该方法；
				this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenFlowControl);
				if ((flowControlTimes2++ % 1000) == 0) {
					log.warn("the queue's messages, span too long, so do flow control, {} {} {}", processQueue.getMaxSpan(), pullRequest, flowControlTimes2);
				}
				return;
			}
		}

		// 6、以PullRequest.messageQueue对象的topic值为参数从RebalanceImpl.subscriptionInner: ConcurrentHashMap<String /* topic */, SubscriptionData>中获取对应的SubscriptionData对象
		final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
		if (null == subscriptionData) {
			// 若该对象为null,由于并发关系，即使找不到订阅关系，也要重试下，防止丢失PullRequest
			this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenException);
			log.warn("find the consumer's subscription failed, {}", pullRequest);
			return;
		}

		final long beginTimestamp = System.currentTimeMillis();

		// 10 消费消息的回调类（PullCallback）
		// 该PullCallback类是DefaultMQPushConsumerImpl.pullMessage (PullRequest pullRequest)方法中的匿名内部类，
		// 采用异步方式拉取消息时，在收到Broker的响应消息之后，回调该方法执行业务调用者的回调逻辑。
		PullCallback pullCallback = new PullCallback() {
			@Override
			public void onSuccess(PullResult pullResult) {
				if (pullResult != null) {
					// **********************************************
					// 1、调用PullAPIWrapper.processPullResult(MessageQueue mq,PullResult pullResult, SubscriptionData subscriptionData)方法处理拉取消息的返回对象PullResult
					pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult, subscriptionData);

					// 2、根据PullResult.status变量的值执行不同的业务逻辑
					switch (pullResult.getPullStatus()) {
						case FOUND :

							// 2.1）该PullRequest对象的nextOffset变量值表示本次消费的开始偏移量，赋值给临时变量prevRequestOffset；
							long prevRequestOffset = pullRequest.getNextOffset();

							// 2.2）取PullResult.nextBeginOffset的值（Broker返回的下一次消费进度的偏移值）赋值给PullRequest.nextOffset变量值；
							pullRequest.setNextOffset(pullResult.getNextBeginOffset());
							long pullRT = System.currentTimeMillis() - beginTimestamp;
							DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(), pullRequest.getMessageQueue().getTopic(), pullRT);

							long firstMsgOffset = Long.MAX_VALUE;

							// 2.3）若PullResult.MsgFoundList列表为空，
							// 则调用DefaultMQPushConsumerImpl.executePullRequestImmediately(PullRequest pullRequest)方法
							// 将该拉取请求对象PullRequest重新延迟放入PullMessageService线程的pullRequestQueue队列中，然后跳出该onSucess方法；否则继续下面的逻辑；
							if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
								DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
							} else {
								firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();

								DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(), pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());

								// 2.4）调用该PullRequest.ProcessQueue对象的putMessage(List<MessageExt> msgs)方法，
								// 将MessageExt列表存入ProcessQueue.msgTreeMap:TreeMap<Long, MessageExt>变量中，
								// 放入此变量的目的是：
								// 第一在顺序消费时从该变量列表中取消息进行消费，
								// 第二可以用此变量中的消息做流控；
								boolean dispathToConsume = processQueue.putMessage(pullResult.getMsgFoundList());

								// ****************回调业务层定义的消费方法***************
								// 2.5）调用ConsumeMessageService.submitConsumeRequest(List <MessageExt> msgs,ProcessQueue processQueue,MessageQueue messageQueue, boolean dispathToConsume)方法，
								// 其中dispathToConsume的值由上一步所得,在顺序消费时使用，为true表示可以消费；
								// 顺序消费 从ProcessQueue对象的TreeMap树形列表中取消息的
								// 并发消费 将从Broker获取的MessageExt消息列表分配到各个 ConsumeRequest 线程中进行并发消费。

								// **********************************************
								// 提交消费请求ConsumeRequest对象，
								// 对于并发消费 该消费请求对象是在ConsumeMessageConcurrentlyService类的内部定义了ConsumeRequest线程类；将此对象提交到ConsumeMessageConcurrentlyService类的线程池中，由该线程完成回调业务层定义的消费方法，
								// 对于顺序消费 该消费请求对象是在ConsumeMessageOrderlyService类的内部定义了ConsumeRequest线程类；将此对象提交到ConsumeMessageOrderlyService类的线程池中，由该线程完成回调业务层定义的消费方法，
								DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(//
										pullResult.getMsgFoundList(), //
										processQueue, //
										pullRequest.getMessageQueue(), //
										dispathToConsume);

								// 2.6）检查拉取消息的间隔时间（DefaultMQPushConsumer.pullInterval，默认为0）
								if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
									// 若大于0，则调用DefaultMQPushConsumerImpl. executePullRequestLater方法，在间隔时间之后再将PullRequest对象放入PullMessageService线程的pullRequestQueue队列中
									DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
								} else {
									// 若不大于0（表示立即再次进行拉取消息），则调用DefaultMQPushConsumerImpl. executePullRequestImmediately方法立即继续下一次拉取消息，从而形成一个循环不间断地拉取消息的过程
									DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
								}
							}

							if (pullResult.getNextBeginOffset() < prevRequestOffset//
									|| firstMsgOffset < prevRequestOffset) {
								log.warn("[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}", //
										pullResult.getNextBeginOffset(), //
										firstMsgOffset, //
										prevRequestOffset);
							}

							break;

						// 3、若PullResult.status=NO_NEW_MSG或者NO_MATCHED_MSG时：
						case NO_NEW_MSG :
							// 3.1）取PullResult.nextBeginOffset的值（Broker返回的下一次消费进度的偏移值）赋值给PullRequest.nextOffset变量值；
							pullRequest.setNextOffset(pullResult.getNextBeginOffset());

							// 3.2）更新消费进度offset
							DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);

							// 3.3）调用DefaultMQPushConsumerImpl.executePullRequestImmediately方法立即继续下一次拉取
							DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
							break;
						case NO_MATCHED_MSG :
							pullRequest.setNextOffset(pullResult.getNextBeginOffset());

							DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);

							DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
							break;

						// 4、若PullResult.status=OFFSET_ILLEGAL
						case OFFSET_ILLEGAL :
							log.warn("the pull request offset illegal, {} {}", //
									pullRequest.toString(), pullResult.toString());

							// 4.1）取PullResult.nextBeginOffset的值（Broker返回的下一次消费进度的偏移值）赋值给PullRequest.nextOffset变量值；
							pullRequest.setNextOffset(pullResult.getNextBeginOffset());

							// 4.2）设置PullRequest.processQueue.dropped等于true，将此该拉取请求作废
							pullRequest.getProcessQueue().setDropped(true);

							// 4.3）创建一个匿名Runnable线程类，
							// 然后调用DefaultMQPushConsumerImpl.executeTaskLater(Runnable r, long timeDelay)方法将该线程类放入PullMessageService.scheduledExecutorService: ScheduledExecutorService调度线程池中，
							// 在10秒钟之后执行该匿名线程类
							DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() {

								@Override
								public void run() {
									try {
										// A）调用OffsetStore.updateOffset(MessageQueue mq, long offset, boolean increaseOnly)方法更新消费进度offset
										DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(), pullRequest.getNextOffset(), false);

										// B）调用OffsetStore.persist(MessageQueue mq)方法：
										// 对于广播模式下offsetStore初始化为LocalFileOffsetStore对象，该对象的persist方法没有处理逻辑；
										// 对于集群模式下offsetStore初始化为RemoteBrokerOffsetStore对象，该对象的persist方法中，
										// =============首先以入参MessageQueue对象为key值从RemoteBrokerOffsetStore.offsetTable: ConcurrentHashMap<MessageQueue, AtomicLong>变量中获取偏移量offset值，
										// =============然后调用updateConsumeOffsetToBroker(MessageQueue mq, long offset)方法向Broker发送UPDATE_CONSUMER_OFFSET请求码的消费进度信息；
										DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());

										// C）以PullRequest对象的messageQueue变量为参数调用RebalanceImpl.removeProcessQueue(MessageQueue mq)方法，
										// 在该方法中，首先从RebalanceImpl.processQueueTable: ConcurrentHashMap<MessageQueue, ProcessQueue>变量中删除MessageQueue记录并返回对应的ProcessQueue对象；
										// 然后该ProcessQueue对象的dropped变量设置为ture；
										// 最后以MessageQueue对象和ProcessQueue对象为参数调用removeUnnecessaryMessageQueue方法删除未使用的消息队列的消费进度
										DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());

										log.warn("fix the pull request offset, {}", pullRequest);
									} catch (Throwable e) {
										log.error("executeTaskLater Exception", e);
									}
								}
							}, 10000);
							break;
						default :
							break;
					}
				}
			}

			// 在发送消息失败或者等待响应超时或者其他异常时回调该onException方法
			@Override
			public void onException(Throwable e) {
				if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
					log.warn("execute the pull request exception", e);
				}

				// 3秒钟之后再将该PullRequest请求重新放入PullMessageService线程的pullRequestQueue队列中；
				DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenException);
			}
		};

		boolean commitOffsetEnable = false;
		long commitOffsetValue = 0L;
		// 7、若消息模型为集群模式（RebalanceImpl.messageModel等于CLUSTERING）
		if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) {

			// 则以PullRequest对象的MessageQueue变量值、type =READ_FROM_MEMORY（从内存中获取消费进度offset值）为参数调用DefaultMQPushConsumerImpl. offsetStore对象（初始化为RemoteBrokerOffsetStore对象）的readOffset(MessageQueue mq, ReadOffsetType type)方法从本地内存中获取消费进度offset值。
			// 集群模式====>以入参MessageQueue对象从RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap <MessageQueue,AtomicLong>变量中获取消费进度偏移量；若该偏移量不为null则返回该值，否则返回-1；
			commitOffsetValue = this.offsetStore.readOffset(pullRequest.getMessageQueue(), ReadOffsetType.READ_FROM_MEMORY);
			if (commitOffsetValue > 0) {
				// 若该offset值大于0 则置临时变量commitOffsetEnable等于true
				/**
				 * 该offset值作为pullKernelImpl方法中的commitOffset参数，在Broker端拉取消息之后根据commitOffsetEnable参数值决定是否用该offset更新消息进度
				 */
				commitOffsetEnable = true;
			}
		}

		String subExpression = null;
		boolean classFilter = false;
		SubscriptionData sd = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
		if (sd != null) {
			// 8、当每次拉取消息之后需要更新订阅关系（由DefaultMQPushConsumer. postSubscriptionWhenPull参数表示，默认为false）
			// 并且以topic值参数从RebalanceImpl.subscriptionInner获取的SubscriptionData对象的classFilterMode等于false（默认为false）
			if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) {
				subExpression = sd.getSubString();
			}

			classFilter = sd.isClassFilterMode();
		}

		// 9、该sysFlag标记的第1个字节置为commitOffsetEnable的值；第2个字节（suspend标记）置为1；第4个字节置为classFilterMode的值；
		int sysFlag = PullSysFlag.buildSysFlag(//
				commitOffsetEnable, // commitOffset
				true, // suspend
				subExpression != null, // subscription
				classFilter // class filter
		);
		try {
			// ************************************************
			// 11、调用底层的拉取消息API接口：
			// PullAPIWrapper.pullKernelImpl(MessageQueue mq, String subExpression, long subVersion,long offset, int maxNums, int sysFlag,long commitOffset,long brokerSuspendMaxTimeMillis, long timeoutMillis, CommunicationMode communicationMode, PullCallback pullCallback)方法进行消息拉取操作
			// 将回调类PullCallback传入该方法中，当采用异步方式拉取消息（详见5.10.2小节）时，在收到响应之后会回调该回调类的方法。
			// PUSH模式和PULL模式下的拉取消息的操作均调用PullAPIWrapper.pullKernelImpl
			this.pullAPIWrapper.pullKernelImpl(//
					pullRequest.getMessageQueue(), // 1
					subExpression, // 2
					subscriptionData.getSubVersion(), // 3
					pullRequest.getNextOffset(), // 4
					this.defaultMQPushConsumer.getPullBatchSize(), // 5
					sysFlag, // 6
					commitOffsetValue, // 7
					BrokerSuspendMaxTimeMillis, // 8
					ConsumerTimeoutMillisWhenSuspend, // 9
					CommunicationMode.ASYNC, // 10
					pullCallback// 11
			);
		} catch (Exception e) {
			log.error("pullKernelImpl exception", e);
			this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenException);
		}
	}

	public void executePullRequestImmediately(final PullRequest pullRequest) {
		this.mQClientFactory.getPullMessageService().executePullRequestImmediately(pullRequest);
	}

	public void executeTaskLater(final Runnable r, final long timeDelay) {
		this.mQClientFactory.getPullMessageService().executeTaskLater(r, timeDelay);
	}

	private void executePullRequestLater(final PullRequest pullRequest, final long timeDelay) {
		this.mQClientFactory.getPullMessageService().executePullRequestLater(pullRequest, timeDelay);
	}

	public boolean isPause() {
		return pause;
	}

	public void setPause(boolean pause) {
		this.pause = pause;
	}

	private void makeSureStateOK() throws MQClientException {
		if (this.serviceState != ServiceState.RUNNING) {
			throw new MQClientException("The consumer service state not OK, "//
					+ this.serviceState//
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null);
		}
	}

	public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end) throws MQClientException, InterruptedException {
		return this.mQClientFactory.getMQAdminImpl().queryMessage(topic, key, maxNum, begin, end);
	}

	public void registerMessageListener(MessageListener messageListener) {
		this.messageListenerInner = messageListener;
	}

	public void resume() {
		this.pause = false;
		log.info("resume this consumer, {}", this.defaultMQPushConsumer.getConsumerGroup());
	}

	public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
		return this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
	}

	public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
		try {
			// 1、根据brokerName获取Broker地址；
			String brokerAddr = (null != brokerName) ? this.mQClientFactory.findBrokerAddressInPublish(brokerName) : RemotingHelper.parseSocketAddressAddr(msg.getStoreHost());

			// 2、调用MQClientAPIImpl.consumerSendMessageBack(String addr, MessageExt msg, String consumerGroup, int delayLevel, long timeoutMillis)方法，
			// 其中delayLevel参数由ConsumeConcurrentlyContext. delayLevelWhenNextConsume可在业务层的回调方法中设置，默认为0（表示由服务器根据重试次数自动叠加）；
			// 2.1、构建ConsumerSendMsgBackRequestHeader对象，
			// =======其中该对象的offset等于MessageExt.commitLogOffset、
			// =======originTopic等于MessageExt.topic、originMsgId等于MessageExt.msgId；
			// 2.2、然后发送CONSUMER_SEND_MSG_BACK请求码的信息给Broker；
			this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg, this.defaultMQPushConsumer.getConsumerGroup(), delayLevel, 5000);
		} catch (Exception e) {

			// 3、如果发送重试消息出现异常，则构建以%RETRY%+consumerGroup为topic值的新Message消息,构建过程如下：
			log.error("sendMessageBack Exception, " + this.defaultMQPushConsumer.getConsumerGroup(), e);

			Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup()), msg.getBody());

			// 3.1)从MessageExt消息的properties属性中获取"ORIGIN_MESSAGE_ID"属性值，
			String originMsgId = MessageAccessor.getOriginMessageId(msg);
			// 若originMsgId为空则以MessageExt消息的msgId来设置新Message信息的properties属性的ORIGIN_MESSAGE_ID属性值，
			// 若有该属性值则以该属性值来设置新Message信息的properties属性的ORIGIN_MESSAGE_ID属性值。
			// 保证同一个消息有多次发送失败能获取到真正消息的msgId；
			MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);

			// 3.2）将MessageExt消息的flag赋值给新Message信息的flag值；
			newMsg.setFlag(msg.getFlag());
			MessageAccessor.setProperties(newMsg, msg.getProperties());

			// 3.3）将MessageExt消息的topic值存入新Message信息的properties属性中"RETRY_TOPIC"属性的值；
			MessageAccessor.putProperty(newMsg, MessageConst.PROPERTY_RETRY_TOPIC, msg.getTopic());

			// 3.4）每次消费重试将MessageExt消息的properties属性中"RECONSUME_TIME"值加1；然后将该值再加3之后存入新Message信息的properties属性中"DELAY"属性的值；
			int reTimes = msg.getReconsumeTimes() + 1;
			MessageAccessor.setReconsumeTime(newMsg, reTimes + "");
			newMsg.setDelayTimeLevel(3 + reTimes);

			// 4、调用在启动Consumer时创建的名为"CLIENT_INNER_PRODUCER"的DefaultMQProducer对象的send(Message msg)方法，
			// 以Consumer内部的消息Producer重发消息（该消息是消费失败且回传给Broker也失败的）；
			this.mQClientFactory.getDefaultMQProducer().send(newMsg);
		}
	}

	public void shutdown() {
		switch (this.serviceState) {
			case CREATE_JUST :
				break;
			case RUNNING :
				this.consumeMessageService.shutdown();
				this.persistConsumerOffset();
				this.mQClientFactory.unregisterConsumer(this.defaultMQPushConsumer.getConsumerGroup());
				this.mQClientFactory.shutdown();
				log.info("the consumer [{}] shutdown OK", this.defaultMQPushConsumer.getConsumerGroup());
				this.rebalanceImpl.destroy();
				this.serviceState = ServiceState.SHUTDOWN_ALREADY;
				break;
			case SHUTDOWN_ALREADY :
				break;
			default :
				break;
		}
	}

	public void start() throws MQClientException {
		// 1、检查DefaultMQPushConsumerImpl.ServiceState的状态（初始化状态为ServiceState.CREATE_JUST）；
		// 只有状态为CREATE_JUST时才启动该Producer；其他状态均不执行启动过程；
		switch (this.serviceState) {
			case CREATE_JUST :
				log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(), this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
				// 2、将DefaultMQPushConsumerImpl.ServiceState置为start_failed,以免客户端同一个进程中重复启动；
				this.serviceState = ServiceState.START_FAILED;

				/**
				 * 3、检查各参数是否正确，规则如下：
				3.1）consumerGroup不能为null，并且不能等于"DEFAULT_CONSUMER"；
				3.2）DefaultMQPushConsumer.messageModel不能为null，默认为MessageModel.CLUSTERING；
				3.3）DefaultMQPushConsumer.ConsumeFromWhere不能为null，默认为ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET；
				3.4）DefaultMQPushConsumer.consumeTimestamp不能为null，并且格式为YYYYMMDDHHMMSS，默认为Consumer启动时的时间前三十分钟；
				3.5）DefaultMQPushConsumer.allocateMessageQueueStrategy不能为null，默认初始化MssageQueue分配策略为AllocateMessageQueueStrategy；
				3.6）DefaultMQPushConsumer.messageListener变量值不能为null；而且messageListener变量值必须是实现MessageListenerConcurrently或MessageListenerOrderly类的实例；
				3.7）最小线程数（consumeThreadMin）和最大线程数（consumeThreadMax）必须在1至1000之间，且最小线程数必须小于最大线程数；
				3.8）DefaultMQPushConsumer.consumeConcurrentlyMaxSpan的值在1至65535之间，默认为2000；
				3.9）DefaultMQPushConsumer.pullThresholdForQueue的值在1至65535之间，默认为1000；
				3.10）DefaultMQPushConsumer.pullInterval在0至65535之间，默认为0；
				3.11）DefaultMQPushConsumer.pullBatchSize在1至1024之间，默认为32；
				 */
				this.checkConfig();
				// 复制订阅关系
				// 4、在应用层对DefaultMQPushConsumer.subscription变量进行的设置，要将此Map变量值转换为SubscriptionData对象
				this.copySubscription();

				// 5、若消息模式为集群模式且实例名（instanceName）等于"DEFAULT"，则取该进程的PID作为该Consumer的实例名（instanceName）；调用java的ManagementFactory.getRuntimeMXBean()方法获取进程的PID；
				if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
					this.defaultMQPushConsumer.changeInstanceNameToPID();
				}

				// 6、与Producer端一样，创建MQClientInstance对象
				this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);

				// 7、初始化rebalance变量
				this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
				this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
				// 设置消费者之间信息分配的策略算法。 初始值可以在consumer中设置consumer.setAllocateMessageQueueStrategy(...);
				// DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerTEST_GROUP");
				// 在初始化的过程中，默认初始化MssageQueue分配策略AllocateMessageQueueAveragely赋值给DefaultMQPushConsumer. allocateMessageQueueStrategy变量
				/*
				 * 可以设置的策略算法： 1、AllocateMessageQueueAveragely() 2、AllocateMessageQueueAveragelyByCircle() 3、AllocateMessageQueueByConfig() 4、AllocateMessageQueueByMachineRoom()
				 */
				this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
				this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);

				// 8、初始化PullAPIWrapper 对象
				this.pullAPIWrapper = new PullAPIWrapper(//
						mQClientFactory, //
						this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
				// 注册FilterMessageHook列表，用于消息的过滤；
				// 该列表可以通过在应用层调用DefaultMQPushConsumerImpl.registerFilterMessageHook(FilterMessageHook hook)方法设置；
				this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);

				// 构建offsetStore消费进度存储对象
				/**
				 * 9、若在应用层通过DefaultMQPushConsumer.setOffsetStore(OffsetStore offsetStore)方法设置了DefaultMQPushConsumer.offsetStore变量，
				 * 则将offsetStore变量赋值给DefaultMQPushConsumerImpl.offsetStore变量；
				 */
				if (this.defaultMQPushConsumer.getOffsetStore() != null) {
					this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
				} else {
					// 若没有设置offsetStore则根据消息模式来设置
					switch (this.defaultMQPushConsumer.getMessageModel()) {
						case BROADCASTING :
							// 若消息模式是广播（BROADCASTING），则初始化LocalFileOffsetStore对象并赋值给DefaultMQPushConsumerImpl.offsetStore变量
							this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
							break;
						case CLUSTERING :
							// 若消息模式是集群（CLUSTERING），则初始化RemoteBrokerOffsetStore对象并赋值给DefaultMQPushConsumerImpl.offsetStore变量
							this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
							break;
						default :
							break;
					}
				}
				// 10、调用DefaultMQPushConsumerImpl.offsetStore:OffsetStore的load方法加载文件；
				// 只有LocalFileOffsetStore类实现了load方法，该方法将本地的offsets.json文件内容加载到LocalFileOffsetStore.offsetTable变量中；
				this.offsetStore.load();

				// 是否消费顺序消息
				if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
					// 11、若DefaultMQPushConsumerImpl.messageListenerInner的值实现了MessageListenerOrderly接口则认为是顺序消费
					this.consumeOrderly = true;

					// *******************顺序消费*************************
					this.consumeMessageService = new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
				} else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
					// 若DefaultMQPushConsumerImpl.messageListenerInner的值实现了MessageListenerConcurrently接口
					this.consumeOrderly = false;

					// *******************并发消费*************************
					this.consumeMessageService = new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
				}
				// 启动消费消息服务
				/**
				 * 12、调用ConsumeMessageService.start方法，
				 * 1、对于ConsumeMessageConcurrentlyService类的start方法没有实现任何业务逻辑；
				 * 2、顺序消费    
				 *    对于ConsumeMessageOrderlyService的start方法的处理逻辑是当消息模式为集群模式时，
				 *    设置定时任务每隔20秒调用一次ConsumeMessageOrderlyService. lockMQPeriodically()方法获取消息队列的分布式锁，
				 *    在该方法内部主要是调用RebalanceImpl.lockAll()方法周期性的获取MessageQueue的分布式锁，
				 *    并将RebalanceImpl.processQueueTable:ConcurrentHashMap <MessageQueue, ProcessQueue>集合中
				 *    获得分布式锁的MessageQueue对象对应的ProcessQueue对象加上本地锁以及记录加锁的时间戳（即该对象的lock=ture、lastLockTimestamp）
				 */
				this.consumeMessageService.start();

				// 向mqClientFactory注册本消费者 将DefaultMQPushConsumerImpl对象在MQClientInstance中注册
				// 13、将DefaultMQPushConsumerImpl对象在MQClientInstance中注册，
				// 以consumerGroup为key值、DefaultMQPushConsumerImpl对象为values值存入MQClientInstance.consumerTable:ConcurrentHashMap<String/* group */, MQConsumerInner>变量中，
				// 若在该变量中已存在该consumerGroup的记录则向应用层抛出MQClientException异常；
				// 说明在一个客户端的一个进程下面启动多个Consumer时consumerGroup名字不能一样，否则无法启动；
				boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
				if (!registerOK) {
					this.serviceState = ServiceState.CREATE_JUST;
					this.consumeMessageService.shutdown();
					throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup() + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL), null);
				}

				// ****************************************启动任务**************************
				// 14、调用MQClientInstance.start()方法，启动MQClientInstance对象
				mQClientFactory.start();
				log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());

				// 15、设置DefaultMQPushConsumerImpl的ServiceState为RUNNING；
				this.serviceState = ServiceState.RUNNING;
				break;
			case RUNNING :
			case START_FAILED :
			case SHUTDOWN_ALREADY :
				throw new MQClientException("The PushConsumer service state not OK, maybe started once, "//
						+ this.serviceState//
						+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null);
			default :
				break;
		}
		// 从namesrv更新topic路由信息
		// 16、遍历DefaultMQPushConsumer.RebalancePushImpl. subscriptionInner:ConcurrentHashMap<String /* topic */, SubscriptionData>队列的Key值（该topic集合是在应用层设置的）,
		// 以每个topic调用MQClientInstance.updateTopicRouteInfoFromNameServer(String topic)方法，更新topic的本地路由信息；
		// 并且在MQClientInstance启动过程中设置的定时任务每隔30秒执行一次该方法；
		this.updateTopicSubscribeInfoWhenSubscriptionChanged();

		// 向所有broker发送心跳信息，（包括订阅关系）
		// 17、立即调用MQClientInstance.sendHeartbeatToAllBrokerWithLock()方法，在该方法中，
		// 第一，向所有在MQClientInstance.brokerAddrTable列表中的Broker发送心跳消息；
		// 第二，向Filter过滤服务器发送REGISTER_MESSAGE_FILTER_CLASS请求码，更新过滤服务器中的Filterclass文件；
		this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();

		// 唤醒Rebalance服务线程
		// 18、调用RebalanceService线程的wakeup方法，唤醒RebalanceService线程服务，立即调用MQClientInstance.doRebalance()方法；
		this.mQClientFactory.rebalanceImmediately();
	}

	private void checkConfig() throws MQClientException {
		Validators.checkGroup(this.defaultMQPushConsumer.getConsumerGroup());

		if (null == this.defaultMQPushConsumer.getConsumerGroup()) {
			throw new MQClientException("consumerGroup is null" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		if (this.defaultMQPushConsumer.getConsumerGroup().equals(MixAll.DEFAULT_CONSUMER_GROUP)) {
			throw new MQClientException("consumerGroup can not equal "//
					+ MixAll.DEFAULT_CONSUMER_GROUP //
					+ ", please specify another one."//
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		if (null == this.defaultMQPushConsumer.getMessageModel()) {
			throw new MQClientException("messageModel is null" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		if (null == this.defaultMQPushConsumer.getConsumeFromWhere()) {
			throw new MQClientException("consumeFromWhere is null" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		Date dt = UtilAll.parseDate(this.defaultMQPushConsumer.getConsumeTimestamp(), UtilAll.yyyyMMddHHmmss);
		if (null == dt) {
			throw new MQClientException("consumeTimestamp is invalid, yyyyMMddHHmmss" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// allocateMessageQueueStrategy
		if (null == this.defaultMQPushConsumer.getAllocateMessageQueueStrategy()) {
			throw new MQClientException("allocateMessageQueueStrategy is null" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// subscription
		if (null == this.defaultMQPushConsumer.getSubscription()) {
			throw new MQClientException("subscription is null" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// messageListener
		if (null == this.defaultMQPushConsumer.getMessageListener()) {
			throw new MQClientException("messageListener is null" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		boolean orderly = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerOrderly;
		boolean concurrently = this.defaultMQPushConsumer.getMessageListener() instanceof MessageListenerConcurrently;
		if (!orderly && !concurrently) {
			throw new MQClientException("messageListener must be instanceof MessageListenerOrderly or MessageListenerConcurrently" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// consumeThreadMin
		if (this.defaultMQPushConsumer.getConsumeThreadMin() < 1 //
				|| this.defaultMQPushConsumer.getConsumeThreadMin() > 1000//
				|| this.defaultMQPushConsumer.getConsumeThreadMin() > this.defaultMQPushConsumer.getConsumeThreadMax()//
		) {
			throw new MQClientException("consumeThreadMin Out of range [1, 1000]" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// consumeThreadMax
		if (this.defaultMQPushConsumer.getConsumeThreadMax() < 1 || this.defaultMQPushConsumer.getConsumeThreadMax() > 1000) {
			throw new MQClientException("consumeThreadMax Out of range [1, 1000]" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// consumeConcurrentlyMaxSpan
		if (this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() < 1 || this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan() > 65535) {
			throw new MQClientException("consumeConcurrentlyMaxSpan Out of range [1, 65535]" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// pullThresholdForQueue
		if (this.defaultMQPushConsumer.getPullThresholdForQueue() < 1 || this.defaultMQPushConsumer.getPullThresholdForQueue() > 65535) {
			throw new MQClientException("pullThresholdForQueue Out of range [1, 65535]" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// pullInterval
		if (this.defaultMQPushConsumer.getPullInterval() < 0 || this.defaultMQPushConsumer.getPullInterval() > 65535) {
			throw new MQClientException("pullInterval Out of range [0, 65535]" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// consumeMessageBatchMaxSize
		if (this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() < 1 || this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize() > 1024) {
			throw new MQClientException("consumeMessageBatchMaxSize Out of range [1, 1024]" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}

		// pullBatchSize
		if (this.defaultMQPushConsumer.getPullBatchSize() < 1 || this.defaultMQPushConsumer.getPullBatchSize() > 1024) {
			throw new MQClientException("pullBatchSize Out of range [1, 1024]" //
					+ FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL), //
					null);
		}
	}

	private void copySubscription() throws MQClientException {
		try {
			Map<String, String> sub = this.defaultMQPushConsumer.getSubscription();
			if (sub != null) {
				// 4.1）遍历DefaultMQPushConsumer.subscription变量
				for (final Map.Entry<String, String> entry : sub.entrySet()) {
					final String topic = entry.getKey();
					final String subString = entry.getValue();
					SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(), //
							topic, subString);
					this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
				}
			}

			if (null == this.messageListenerInner) {
				/**
				 * 4.2）若DefaultMQPushConsumerImpl.messageListenerInner为null，则将DefaultMQPushConsumer.messageListener变量赋值给它；
				 * ******************当应用层通过调用DefaultMQPushConsumer.setMessageListener(MessageListener messageListener)方法来设置回调类时，*************************
				 * 就要在此给DefaultMQPushConsumerImpl.messageListenerInner赋值；
				 */
				this.messageListenerInner = this.defaultMQPushConsumer.getMessageListener();
			}

			switch (this.defaultMQPushConsumer.getMessageModel()) {
				case BROADCASTING :
					break;

				case CLUSTERING :
					/**
					 * 4.3）若消息模式为集群（DefaultMQPushConsumer.messageModel=CLUSTERING），
					 * 则生成以 "%RETRY%"+consumerGroup名为topic、"*"为subExpression的SubscriptionData对象，
					 * 并存入DefaultMQPushConsumer. RebalancePushImpl.subscriptionInner变量中；
					 */
					final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
					SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(), //
							retryTopic, SubscriptionData.SUB_ALL);
					this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
					break;
				default :
					break;
			}
		} catch (Exception e) {
			throw new MQClientException("subscription exception", e);
		}
	}

	public MessageListener getMessageListenerInner() {
		return messageListenerInner;
	}

	private void updateTopicSubscribeInfoWhenSubscriptionChanged() {
		Map<String/* topic */, SubscriptionData> subTable = this.getSubscriptionInner(); // DefaultMQPushConsumer.RebalancePushImpl. subscriptionInner:ConcurrentHashMap<String /* topic */, SubscriptionData>
		if (subTable != null) {
			for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
				final String topic = entry.getKey();
				this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
			}
		}
	}

	public void subscribe(String topic, String subExpression) throws MQClientException {
		try {
			// 构造SubscriptionData对象并返回
			SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(), //
					topic, subExpression);
			// 4.2）以topic和返回的SubscriptionData对象为K-V值存入DefaultMQPushConsumer.RebalancePushImpl.subscriptionInner:ConcurrentHashMap<String /* topic */, SubscriptionData>变量中；
			this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
			if (this.mQClientFactory != null) {
				// 4.3）在该方法中，第一，向所有在MQClientInstance.brokerAddrTable列表中的Broker发送心跳消息；
				// 第二，向Filter过滤服务器发送REGISTER_MESSAGE_FILTER_CLASS请求码，更新过滤服务器中的Filterclass文件；
				this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
			}
		} catch (Exception e) {
			throw new MQClientException("subscription exception", e);
		}
	}

	public void subscribe(String topic, String fullClassName, String filterClassSource) throws MQClientException {
		try {
			SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(), //
					topic, "*");
			subscriptionData.setSubString(fullClassName);
			subscriptionData.setClassFilterMode(true);
			subscriptionData.setFilterClassSource(filterClassSource);
			this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
			if (this.mQClientFactory != null) {
				this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
			}

		} catch (Exception e) {
			throw new MQClientException("subscription exception", e);
		}
	}

	public void suspend() {
		this.pause = true;
		log.info("suspend this consumer, {}", this.defaultMQPushConsumer.getConsumerGroup());
	}

	public void unsubscribe(String topic) {
		this.rebalanceImpl.getSubscriptionInner().remove(topic);
	}

	public void updateConsumeOffset(MessageQueue mq, long offset) {
		this.offsetStore.updateOffset(mq, offset, false);
	}

	public void updateCorePoolSize(int corePoolSize) {
		this.consumeMessageService.updateCorePoolSize(corePoolSize);
	}

	public MessageExt viewMessage(String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
		return this.mQClientFactory.getMQAdminImpl().viewMessage(msgId);
	}

	public RebalanceImpl getRebalanceImpl() {
		return rebalanceImpl;
	}

	public boolean isConsumeOrderly() {
		return consumeOrderly;
	}

	public void setConsumeOrderly(boolean consumeOrderly) {
		this.consumeOrderly = consumeOrderly;
	}

	@Override
	public boolean isUnitMode() {
		return this.defaultMQPushConsumer.isUnitMode();
	}

	public void resetOffsetByTimeStamp(long timeStamp) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
		for (String topic : rebalanceImpl.getSubscriptionInner().keySet()) {
			Set<MessageQueue> mqs = rebalanceImpl.getTopicSubscribeInfoTable().get(topic);
			Map<MessageQueue, Long> offsetTable = new HashMap<MessageQueue, Long>();
			if (mqs != null) {
				for (MessageQueue mq : mqs) {
					long offset = searchOffset(mq, timeStamp);
					offsetTable.put(mq, offset);
				}
				this.mQClientFactory.resetOffset(topic, groupName(), offsetTable);
			}
		}
	}

	public MQClientInstance getmQClientFactory() {
		return mQClientFactory;
	}

	public void setmQClientFactory(MQClientInstance mQClientFactory) {
		this.mQClientFactory = mQClientFactory;
	}

	public ServiceState getServiceState() {
		return serviceState;
	}

	public void setServiceState(ServiceState serviceState) {
		this.serviceState = serviceState;
	}

	private long computeAccumulationTotal() {
		long msgAccTotal = 0;
		ConcurrentHashMap<MessageQueue, ProcessQueue> processQueueTable = this.rebalanceImpl.getProcessQueueTable();
		Iterator<Entry<MessageQueue, ProcessQueue>> it = processQueueTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<MessageQueue, ProcessQueue> next = it.next();
			ProcessQueue value = next.getValue();
			msgAccTotal += value.getMsgAccCnt();
		}

		return msgAccTotal;
	}

	public void adjustThreadPool() {
		// 首先汇总DefaultMQPushConsumerImpl.RebalanceImpl.processQueueTable中所有ProcessQueue对象的msgAccCnt值（即表示Broker端未消费的逻辑队列大小）；
		long computeAccTotal = this.computeAccumulationTotal();
		// 由DefaultMQPushConsumer.adjustThreadPoolNumsThreshold参数设置阀值，默认为100000；
		long adjustThreadPoolNumsThreshold = this.defaultMQPushConsumer.getAdjustThreadPoolNumsThreshold();

		long incThreshold = (long) (adjustThreadPoolNumsThreshold * 1.0);

		long decThreshold = (long) (adjustThreadPoolNumsThreshold * 0.8);

		// 若汇总的msgAccCnt值大于adjustThreadPoolNumsThreshold值，并且ConsumeMessageConcurrentlyService.consumeExecutor的当前线程数量小于最大线程数，则将当前线程数量加1；
		if (computeAccTotal >= incThreshold) {
			// ConsumeMessageService类有两个子类ConsumeMessageConcurrentlyService和ConsumeMessageOrderlyService，分别表示并发消费和顺序消费两类情况，这两个之类的adjustThreadPool方法是一样的
			this.consumeMessageService.incCorePoolSize();
		}

		// 若汇总的msgAccCnt值小于adjustThreadPoolNumsThreshold*0.8，并且ConsumeMessageConcurrentlyService.consumeExecutor的当前线程数量大于最小线程数，则将当前线程数量减1；
		if (computeAccTotal < decThreshold) {
			this.consumeMessageService.decCorePoolSize();
		}
	}

	@Override
	public ConsumerRunningInfo consumerRunningInfo() {
		ConsumerRunningInfo info = new ConsumerRunningInfo();

		Properties prop = MixAll.object2Properties(this.defaultMQPushConsumer);

		prop.put(ConsumerRunningInfo.PROP_CONSUME_ORDERLY, String.valueOf(this.consumeOrderly));
		prop.put(ConsumerRunningInfo.PROP_THREADPOOL_CORE_SIZE, String.valueOf(this.consumeMessageService.getCorePoolSize()));
		prop.put(ConsumerRunningInfo.PROP_CONSUMER_START_TIMESTAMP, String.valueOf(this.consumerStartTimestamp));

		info.setProperties(prop);

		Set<SubscriptionData> subSet = this.subscriptions();
		info.getSubscriptionSet().addAll(subSet);

		Iterator<Entry<MessageQueue, ProcessQueue>> it = this.rebalanceImpl.getProcessQueueTable().entrySet().iterator();
		while (it.hasNext()) {
			Entry<MessageQueue, ProcessQueue> next = it.next();
			MessageQueue mq = next.getKey();
			ProcessQueue pq = next.getValue();

			ProcessQueueInfo pqinfo = new ProcessQueueInfo();
			pqinfo.setCommitOffset(this.offsetStore.readOffset(mq, ReadOffsetType.MEMORY_FIRST_THEN_STORE));
			pq.fillProcessQueueInfo(pqinfo);
			info.getMqTable().put(mq, pqinfo);
		}

		for (SubscriptionData sd : subSet) {
			ConsumeStatus consumeStatus = this.mQClientFactory.getConsumerStatsManager().consumeStatus(this.groupName(), sd.getTopic());
			info.getStatusTable().put(sd.getTopic(), consumeStatus);
		}

		return info;
	}

	public ConsumerStatsManager getConsumerStatsManager() {
		return this.mQClientFactory.getConsumerStatsManager();
	}

	public Set<QueueTimeSpan> queryConsumeTimeSpan(final String topic) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
		Set<QueueTimeSpan> queueTimeSpan = new HashSet<QueueTimeSpan>();
		TopicRouteData routeData = this.mQClientFactory.getMQClientAPIImpl().getTopicRouteInfoFromNameServer(topic, 3000);
		for (BrokerData brokerData : routeData.getBrokerDatas()) {
			String addr = brokerData.selectBrokerAddr();
			queueTimeSpan.addAll(this.mQClientFactory.getMQClientAPIImpl().queryConsumeTimeSpan(addr, topic, groupName(), 3000l));
		}

		return queueTimeSpan;
	}

	public ConsumeMessageService getConsumeMessageService() {
		return consumeMessageService;
	}

	public void setConsumeMessageService(ConsumeMessageService consumeMessageService) {
		this.consumeMessageService = consumeMessageService;

	}
}
